package com.example.mqdemo.listener;

import cn.hutool.core.util.StrUtil;
import com.example.mqdemo.bean.MsgData;
import com.example.mqdemo.common.QueueConstant;
import com.example.mqdemo.util.MsgUtil;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.connection.PublisherCallbackChannel;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;

import java.io.IOException;
import java.util.Map;

@Slf4j
@Configuration
public class RabbitListenerConfig {

    @Autowired
    MsgUtil msgUtil;

    @RabbitListener(queues = {QueueConstant.orderQueueName})
    @RabbitHandler
    public void orderDirectListener(Message msg, Channel channel, @Headers Map<String, Object> headers, @Payload String body) throws IOException {
        String correlationId = msg.getMessageProperties().getCorrelationId();
        if (StrUtil.isBlank(correlationId)) {
            correlationId = StrUtil.toString(headers.get(PublisherCallbackChannel.RETURNED_MESSAGE_CORRELATION_KEY));
        }
        try {
//            log.info("Listener接收到消息:{}", new String(msg.getBody()));
            log.info("Listener(创建订单)接收到消息:{}", body);
        } catch (Exception e) {

            record(correlationId, "error", e.getMessage(), msg.getMessageProperties().getDeliveryTag(), "create");

            // 自己实现消息重试机制
            /**
             * deliveryTag:该消息的index
             * multiple：是否批量.true:将一次性拒绝所有小于deliveryTag的消息
             * requeue：被拒绝的是否重新入队列 注意：如果设置为true ，则会添加在队列的末端
             */
            channel.basicNack(msg.getMessageProperties().getDeliveryTag(), false, true);
        } finally {

            record(correlationId, "success", "ack", msg.getMessageProperties().getDeliveryTag(), "create");

            /**
             * prefetchSize：消息的大小
             * prefetchCount：会告诉RabbitMQ不要同时给一个消费者推送多于N个消息，即一旦有N个消息还没有ack，则该consumer将block掉，直到有消息ack
             * global：是否将上面设置应用于channel，简单点说，就是上面限制是channel级别(true)的还是consumer级别(false)
             */
            // channel.basicQos(0, 1, true);

            /**
             * deliveryTag：该消息的index
             * multiple：是否批量处理.true:将一次性ack所有小于deliveryTag的消息
             */
            channel.basicAck(msg.getMessageProperties().getDeliveryTag(), false);
        }
    }

    @RabbitListener(queues = {QueueConstant.orderCancelQueueName})
    @RabbitHandler
    public void orderCancelDirectListener(Message msg, Channel channel, @Headers Map<String, Object> headers, @Payload String body) throws IOException {
        String correlationId = msg.getMessageProperties().getCorrelationId();
        if (StrUtil.isBlank(correlationId)) {
            correlationId = StrUtil.toString(headers.get(PublisherCallbackChannel.RETURNED_MESSAGE_CORRELATION_KEY));
        }
        try {
//            log.info("Listener接收到消息:{}", new String(msg.getBody()));
            log.info("Listener(取消订单)接收到消息:{}", body);
        } catch (Exception e) {

            record(correlationId, "error", e.getMessage(), msg.getMessageProperties().getDeliveryTag(), "Cancel");

            // 自己实现消息重试机制
            /**
             * deliveryTag:该消息的index
             * multiple：是否批量.true:将一次性拒绝所有小于deliveryTag的消息
             * requeue：被拒绝的是否重新入队列 注意：如果设置为true ，则会添加在队列的末端
             */
            channel.basicNack(msg.getMessageProperties().getDeliveryTag(), false, true);
        } finally {

            record(correlationId, "success", "ack", msg.getMessageProperties().getDeliveryTag(), "Cancel");

            /**
             * prefetchSize：消息的大小
             * prefetchCount：会告诉RabbitMQ不要同时给一个消费者推送多于N个消息，即一旦有N个消息还没有ack，则该consumer将block掉，直到有消息ack
             * global：是否将上面设置应用于channel，简单点说，就是上面限制是channel级别(true)的还是consumer级别(false)
             */
            // channel.basicQos(0, 1, true);

            /**
             * deliveryTag：该消息的index
             * multiple：是否批量处理.true:将一次性ack所有小于deliveryTag的消息
             */
            channel.basicAck(msg.getMessageProperties().getDeliveryTag(), false);
        }
    }

    private void record(String id, String status, String msg, long tag, String type) {
        MsgData msgData = msgUtil.getMsgData(id);
        if (msgData == null) {
            return;
        }
        msgData.setStatus(status);
        msgData.setMsg(msg);
        msgData.setTag(tag);
        msgData.setType(type);
        msgUtil.setData(id, msgData);
    }
}
